import guru.kafka.test.RandomPartitioner;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * Created by xufei on 16/10/13.
 */
public class TestProducer {

    public static Properties getProperties() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "127.0.0.1:9092");
        properties.put("metadata.broker.list", "127.0.0.1:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("serializer.class", "kafka.serializer.StringEncoder");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("request.required.acks", "0");

        properties.put("partitions.class", RandomPartitioner.class.getName());

        return properties;
    }

    public static final void main(String[] args) {
        Properties properties = getProperties();

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        // Producer<String, String> producer = new Producer<String, String>(new ProducerConfig(properties));
        for (int iCount = 0; iCount < 100; iCount++) {
            String message = "My Test Message No " + iCount;

            System.out.println(message);

            ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic1", message);
            // KeyedMessage<String, String> record = new KeyedMessage<String, String>("topic1", "11", message);
            producer.send(record);
        }
        producer.close();
    }
}
